package org.example.dataset;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.CloseableIterator;
import org.apache.flink.util.Collector;
import org.example.Main;
import org.example.domain.Result;
import org.springframework.boot.SpringApplication;
import org.springframework.stereotype.Service;

import java.io.ByteArrayOutputStream;
import java.util.*;


public class DataHandler {
//    public void read(List<String>){
//        StreamExecutionEnvironment env = new StreamExecutionEnvironment();
//        DataStreamSource<String> lines = env.readTextFile("./a.txt");
//    }

    public static void handleResult(List<Result> res){

        StreamExecutionEnvironment env = new StreamExecutionEnvironment();
        DataStreamSource<Result> resultDataStreamSource = env.fromCollection(res);
        SingleOutputStreamOperator<Tuple2<String, Double>> tuple2SingleOutputStreamOperator = resultDataStreamSource.flatMap(new FlatMapFunction<Result, Tuple2<String, Double>>() {
            @Override
            public void flatMap(Result result, Collector<Tuple2<String, Double>> collector) throws Exception {
//                System.out.println(result);
                collector.collect(new Tuple2<>(result.getCode(), result.getScore()));
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Double>> sum = tuple2SingleOutputStreamOperator.keyBy(tuple -> tuple.f0).sum(1);
//        sum.print();
    }



    public static List<String> handle() throws Exception {
        List<Result> res = Arrays.asList(
                new Result("s1", 0.85),
                new Result("s2", 0.5),
                new Result("s3", 0.66)
        );
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        DataStreamSource<Result> resultDataStreamSource = env.fromCollection(res);



        SingleOutputStreamOperator<Tuple2<String, Double>> tuple2SingleOutputStreamOperator = resultDataStreamSource.flatMap(new FlatMapFunction<Result, Tuple2<String, Double>>() {
            @Override
            public void flatMap(Result result, Collector<Tuple2<String, Double>> collector) throws Exception {
//                System.out.println(result);
                collector.collect(new Tuple2<>(result.getCode(), result.getScore()));
            }
        });


        KeyedStream<Tuple2<String, Double>, Tuple> tuple2TupleKeyedStream = tuple2SingleOutputStreamOperator.keyBy(0);

//        tuple2TupleKeyedStream.writeAsText("file/out1").setParallelism(1);
        SingleOutputStreamOperator<String> stringSingleOutputStreamOperator = tuple2TupleKeyedStream.flatMap(new FlatMapFunction<Tuple2<String, Double>, String>() {
            @Override
            public void flatMap(Tuple2<String, Double> stringDoubleTuple2, Collector<String> collector) throws Exception {
                collector.collect(stringDoubleTuple2.f0 + stringDoubleTuple2.f1.toString());
            }
        });

//        ByteArrayOutputStream buffer = new ByteArrayOutputStream();
//        int nRead;
//        byte[] data = new byte[1024];
//        while ((nRead = stringSingleOutputStreamOperator.read(data, 0, data.length)) != -1) {
//            buffer.write(data, 0, nRead);
//        }

        List<String> s= (List<String>) stringSingleOutputStreamOperator.executeAndCollect();
//        stringSingleOutputStreamOperator.print();


        env.execute();
        return s;
    }


    public static List<Tuple2<String, Double>> handleDataSet() throws Exception {
        List<Result> res = Arrays.asList(
                new Result("s1", 0.85),
                new Result("s2", 0.5),
                new Result("s3", 0.66)
        );
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<Result> datas = executionEnvironment.fromCollection(res);

        FlatMapOperator<Result, Tuple2<String, Double>> resultTuple2FlatMapOperator = datas.flatMap(new FlatMapFunction<Result, Tuple2<String, Double>>() {
            @Override
            public void flatMap(Result result, Collector<Tuple2<String, Double>> collector) throws Exception {
//                System.out.println(result);
                collector.collect(new Tuple2<>(result.getCode(), result.getScore()));
            }
        });

        List<Tuple2<String, Double>> collect = resultTuple2FlatMapOperator.collect();
        return collect;
    }



    public static void main(String[] args) throws Exception {
        List<Tuple2<String, Double>> handle = handleDataSet();
        for (Tuple2<String, Double> stringDoubleTuple2 : handle) {
            System.out.println(stringDoubleTuple2);
        }
    }

}

